package com.es.hotel.listen;

import com.alibaba.fastjson.JSON;
import com.es.hotel.constants.HotelMqConstants;
import com.es.hotel.constants.MessageQueueIdempotentHandler;
import com.es.hotel.pojo.Hotel;
import com.es.hotel.pojo.HotelDoc;
import com.es.hotel.service.impl.HotelService;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.connection.stream.StreamReadOptions;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletionException;

//酒店监听器
@Component
public class HotelListener {

    @Autowired
    private HotelService hotelService;

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    @Autowired
    private MessageQueueIdempotentHandler messageQueueIdempotentHandler;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @PostConstruct
    public void startConsuming() {
        new Thread(this::consumeStreamBlocking).start();  // 启动一个新线程来执行阻塞操作
    }

    //阻塞式消费消息
    public void consumeStreamBlocking(){
        StreamOperations<String, String, String> streamOps = stringRedisTemplate.opsForStream();
        while (true) {
            List<MapRecord<String, String, String>> messages = streamOps.read(
                    StreamReadOptions.empty().block(Duration.ofSeconds(10)).count(1),  // 从最新的位置读取
                    (StreamOffset<String>) StreamOffset.latest(HotelMqConstants.HOTEL_STREAM_KEY)  // 阻塞等待 10 秒
                        );
            if (messages != null && !messages.isEmpty()) {
                for (MapRecord<String, String, String> message : messages) {
                    insertHotel(message);
                }
            }
        }
    }

    //新增和更新
    public void insertHotel(MapRecord<String, String, String> message) {
        String stream = message.getStream();  // 获取消息所属的 Redis Stream
        RecordId id = message.getId();        // 获取消息的唯一标识符
        try {
            // 消息幂等性处理，检查消息是否已经被消费过
            if (messageQueueIdempotentHandler.isMessageBeingConsumed(id.toString())) {
                // 判断当前的消息是否已经完全处理
                if (messageQueueIdempotentHandler.isAccomplish(id.toString())) {
                    return;  // 已经完成处理，直接返回
                }
                throw new RuntimeException("消息未完成，需要消息队列重试");
            }
            // 获取消息内容
            Map<String, String> map = message.getValue();
            Long hotelId = JSON.parseObject(map.get("hotelId"), Long.class);
            // 获取酒店信息并进行处理
            Hotel hotel = hotelService.getById(hotelId);
            HotelDoc hotelDoc = new HotelDoc(hotel);
            // 将酒店数据索引到 Elasticsearch 中
            IndexRequest request = new IndexRequest("hotel");
            request.source(JSON.toJSONString(hotelDoc), XContentType.JSON);
            restHighLevelClient.index(request, RequestOptions.DEFAULT);
            // 删除已经消费的消息，避免重复消费
            stringRedisTemplate.opsForStream().delete(Objects.requireNonNull(stream), id.getValue());
        } catch (IOException e) {
            // 如果消费失败，删除 Redis 中的消息处理标识
            messageQueueIdempotentHandler.delMessageProcessed(id.toString());
            throw new RuntimeException("新增酒店数据失败: " + e);
        }
        // 标记消息已经处理完成
        messageQueueIdempotentHandler.setAccomplish(id.toString());
    }



}
